package com.changdu.seatunnel.admin.config;

import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import com.changdu.seatunnel.admin.handlers.DingRobotHandler;
import com.changdu.seatunnel.admin.handlers.JobStateScanHandler;
import com.changdu.seatunnel.admin.mapper.DataInstanceMapper;
import com.changdu.seatunnel.admin.service.DataSourceService;
import com.changdu.seatunnel.admin.service.KafkaCreater;
import com.changdu.seatunnel.admin.service.SyncTaskService;
import com.changdu.seatunnel.admin.service.impl.DatabaseDatasourceConfigServiceImpl;
import com.changdu.seatunnel.admin.service.impl.ExternalDatasourceConfigServiceImpl;

@Configuration
@EnableConfigurationProperties(SeatunnelQueryProperty.class)
public class SyncDataConfiguration {
	
	public DataSourceService databaseDatasourceConfigService(DataInstanceMapper datasourceConfigMapper) {
		return new DatabaseDatasourceConfigServiceImpl(datasourceConfigMapper);
	}
	
	public DataSourceService externalDatasourceConfigService(DataInstanceMapper datasourceConfigMapper, SeatunnelQueryProperty seatunnelQueryProperty) {
		return new ExternalDatasourceConfigServiceImpl(datasourceConfigMapper, seatunnelQueryProperty);
	}
	
	@Bean
	public DataSourceService getDatasourceConfigService(DataInstanceMapper datasourceConfigMapper, SeatunnelQueryProperty seatunnelQueryProperty) {
		if ("mysql".equalsIgnoreCase(seatunnelQueryProperty.getType())) {
			return databaseDatasourceConfigService(datasourceConfigMapper);
		}
		return externalDatasourceConfigService(datasourceConfigMapper, seatunnelQueryProperty);
	}
	
	/**
	 * kafka client
	 * @return
	 */
	@Bean
	public KafkaCreater kafkaClient() {
		return new KafkaCreater();
	}
	
	/**
	 * 钉钉实例
	 * @param seatunnelQueryProperty
	 * @return
	 */
	@Bean
	public DingRobotHandler DingRobotHandler(SeatunnelQueryProperty seatunnelQueryProperty) {
		return new DingRobotHandler(seatunnelQueryProperty.getDing());
	}
	
	@Bean
	public JobStateScanHandler jobStateScanHandler(DingRobotHandler DingRobotHandler, SyncTaskService syncTaskService, 
			KafkaCreater kafkaClient, SeatunnelQueryProperty seatunnelQueryProperty) {
		return new JobStateScanHandler(seatunnelQueryProperty, DingRobotHandler, syncTaskService, kafkaClient);
	}
	
}
